package test.threads.queue.stressed;

import java.util.List;
import java.util.concurrent.Executor;

import com.google.common.collect.Lists;
import com.lmax.disruptor.*;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import test.helpers.Config;
import test.threads.queue.common.IUsefulWorkEmulator;

import static com.google.common.base.Preconditions.checkArgument;

/**
 * (source) -> 0 -> 1 -> 2 -> ... -> N
 *
 * @author cheremin
 * @since 27.02.12,  15:00
 */
public class DxNTask<E> extends AbstractTask<E> {
    private static final Log log = LogFactory.getLog( DxNTask.class );

    private final WaitStrategy waitStrategy;

    private final boolean publishInBatch;
    private final int batchSize;
    private final boolean subscribeInBatch;

    private final int nodes;

    public DxNTask( final Config config ) throws ClassNotFoundException, InstantiationException, IllegalAccessException {
        super( config );

        this.waitStrategy = config.newInstanceByClass(
                "disruptor.wait-strategy",
                new BusySpinWaitStrategy()
        );

        this.nodes = config.getAsInt( "pipeline.nodes" );

        this.publishInBatch = config.getAsBoolean( "disruptor.publish-in-batch", false );
        if ( this.publishInBatch ) {
            batchSize = config.getAsInt( "disruptor.batch-size", 1 );
        } else {
            batchSize = -1;
        }
        this.subscribeInBatch = config.getAsBoolean( "disruptor.subscribe-in-batch", true );
    }


    private RingBuffer<E> ringBuffer;
    private EventProcessor[] processors;

    @Override
    public int workersRequired() {
        return nodes;
    }

    private SequenceBarrier lastSequenceBarrier;
    private final Sequence lastSequence = new Sequence( Sequencer.INITIAL_CURSOR_VALUE );
    private IUsefulWorkEmulator<E> lastEmulator;

    @Override
    protected EventEnqueuer _initialize( final Executor threadPool,
                                         final IUsefulWorkEmulator<E>[] emulators,
                                         final EventFactory<E> eventFactory ) {
        final ClaimStrategy claimStrategy = new SingleThreadedClaimStrategy( bufferSize() );

        ringBuffer = new RingBuffer<E>(
                eventFactory,
                claimStrategy,
                waitStrategy
        );

        SequenceBarrier previousBarrier = ringBuffer.newBarrier();
        final List<EventProcessor> processors = Lists.newArrayList();
        for ( int i = 0; i < nodes - 1; i++ ) {
            final IUsefulWorkEmulator<E> emulator = emulators[i];
            final PassThroughHandler<E> handler = new PassThroughHandler<E>( emulator );
            final BatchEventProcessor<E> processor = new BatchEventProcessor<E>(
                    ringBuffer,
                    previousBarrier,
                    handler
            );
            previousBarrier = ringBuffer.newBarrier( processor.getSequence() );
            processors.add( processor );
        }

        this.processors = processors.toArray( new EventProcessor[0] );

        lastSequenceBarrier = previousBarrier;
        lastEmulator = emulators[emulators.length - 1];

        ringBuffer.setGatingSequences( lastSequence );

        for ( final EventProcessor processor : processors ) {
            threadPool.execute( processor );
        }

        return new Enqueuer( eventsInBatch(), batchSize, eventFactory, ringBuffer );
    }

    @Override
    protected long dequeueBatchEvents( final int eventsInBatch ) throws Exception {
        final long seq = lastSequence.get();
        final long maxSequence = seq + eventsInBatch;


        long nextSequence = lastSequence.get() + 1L;
        while ( nextSequence <= maxSequence ) {
            final long availableSequence = Math.min(
                    lastSequenceBarrier.waitFor( nextSequence ),
                    maxSequence
            );
            while ( nextSequence <= availableSequence ) {
                final E event = ringBuffer.get( nextSequence );
                lastEmulator.spendCPUTime( event );

                nextSequence++;
            }
        }
        lastSequence.set( nextSequence - 1L );

        return 0;
    }


    @Override
    public void terminate() {
        super.terminate();

        for ( final EventProcessor processor : processors ) {
            processor.halt();
        }
    }

    @Override
    public String toString() {
        return String.format(
                "%s[%s][%s]",
                super.toString(),
                waitStrategy.getClass().getSimpleName(),
                ( publishInBatch ) ? ( "batching" ) : ( "no-batching" )
        );
    }

    private static class PassThroughHandler<E> implements EventHandler<E> {

        private final IUsefulWorkEmulator<E> emulator;

        public PassThroughHandler( final IUsefulWorkEmulator<E> emulator ) {
            this.emulator = emulator;
        }

        @Override
        public void onEvent( final E event,
                             final long sequence,
                             final boolean endOfBatch ) throws Exception {
            emulator.spendCPUTime( event );
        }
    }

    private class Enqueuer extends EventEnqueuer<E> {
        private final RingBuffer<E> ringBuffer;

        private final BatchDescriptor batch;
        private final int batchesInBatch;

        private Enqueuer( final int eventsInBatch,
                          final int batchSize,
                          final EventFactory<E> factory,
                          final RingBuffer<E> ringBuffer ) {
            super( eventsInBatch, factory );
            this.ringBuffer = ringBuffer;

            checkArgument(
                    eventsInBatch % batchSize == 0,
                    "eventsInBatch(%s) % batchSize(%s) must be 0",
                    eventsInBatch, batchSize
            );
            batch = ringBuffer.newBatchDescriptor( batchSize );

            batchesInBatch = eventsInBatch / batchSize;
        }

        @Override
        protected void enqueueEventsBatch( final int eventsInBatch,
                                           final EventFactory<E> factory ) throws Exception {
            if ( publishInBatch ) {
                int i = 0;
                for ( int batchNo = 0; batchNo < batchesInBatch; batchNo++ ) {
                    ringBuffer.next( batch );
                    for ( int eventNo = 0; eventNo < batch.getSize(); eventNo++ ) {
                        final long sequence = batch.getStart() + eventNo;
//                        final LongValueMessage event = ringBuffer.get( sequence );
//                        event.value = i;
                        i++;
                    }
                    ringBuffer.publish( batch );
                }
            } else {
                for ( int i = 0; i < eventsInBatch; i++ ) {
                    final long nextSequence = ringBuffer.next();
                    ringBuffer.get( nextSequence );
                    ringBuffer.publish( nextSequence );
                }
            }
        }
    }
}
